// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#pragma once

#include <algorithm>
#include <memory>
#include <string>
#include <utility>

#include <fmt/format.h>
#include <melon/format.h>
#include <melon/range.h>
#include <melon/container/f14_map.h>

#include <pollux/buffer/buffer.h>
#include <pollux/common/base/bit_util.h>
#include <pollux/common/base/compare_flags.h>
#include <pollux/common/base/exceptions.h>
#include <pollux/common/base/nulls.h>
#include <pollux/type/type.h>
#include <pollux/type/variant.h>
#include <pollux/vector/builder_type_utils.h>
#include <pollux/vector/selectivity_vector.h>
#include <pollux/vector/type_aliases.h>
#include <pollux/vector/vector_encoding.h>
#include <pollux/vector/vector_util.h>

namespace kumo::pollux {
    template<typename T>
    class SimpleVector;

    template<typename T>
    class FlatVector;

    class VectorPool;
    class BaseVector;
    using VectorPtr = std::shared_ptr<BaseVector>;

    /// Set of options that validate() accepts.
    struct VectorValidateOptions {
        /// If set to true then an unloaded lazy vector is loaded and validate is
        /// called on the loaded vector. NOTE: loading a vector is non-trivial and
        /// can modify how it's handled by downstream code-paths.
        bool loadLazy = false;

        /// Any optional checks you want to execute on the vector.
        std::function<void(const BaseVector &)> callback;
    };

    class DecodedVector;

    /// Base class for all columnar-based vectors of any type.
    class BaseVector {
    public:
        BaseVector(const BaseVector &) = delete;

        BaseVector &operator=(const BaseVector &) = delete;

        static constexpr uint64_t kNullHash = 1;

        BaseVector(
            pollux::memory::MemoryPool *pool,
            TypePtr type,
            VectorEncoding::Simple encoding,
            BufferPtr nulls,
            size_t length,
            std::optional<vector_size_t> distinctValueCount = std::nullopt,
            std::optional<vector_size_t> nullCount = std::nullopt,
            std::optional<ByteCount> representedByteCount = std::nullopt,
            std::optional<ByteCount> storageByteCount = std::nullopt);

        virtual ~BaseVector() = default;

        VectorEncoding::Simple encoding() const {
            return encoding_;
        }

        inline bool is_lazy() const {
            return encoding() == VectorEncoding::Simple::LAZY;
        }

        /// Returns false if vector has no nulls. Return true if vector may have
        /// nulls.
        virtual bool may_have_nulls() const {
            return rawNulls_;
        }

        /// Returns false if this vector and all of its children have no nulls.
        /// Returns true if this vector or any of its children may have nulls.
        virtual bool may_have_nulls_recursive() const {
            return may_have_nulls();
        }

        inline bool is_index_in_range(vector_size_t index) const {
            // This compiles better than index >= 0 && index < length_.
            return static_cast<uint32_t>(index) < length_;
        }

        template<typename T>
        T *as() {
            static_assert(std::is_base_of_v<BaseVector, T>);
            return dynamic_cast<T *>(this);
        }

        template<typename T>
        const T *as() const {
            static_assert(std::is_base_of_v<BaseVector, T>);
            return dynamic_cast<const T *>(this);
        }

        /// Use when the type of 'this' is already known. dynamic_cast() is slow.
        template<typename T>
        T *as_unchecked() {
            static_assert(std::is_base_of_v<BaseVector, T>);
            POLLUX_DCHECK_NOT_NULL(
                dynamic_cast<const T*>(this),
                "Wrong type cast expected {}, but got {}",
                typeid(T).name(),
                typeid(*this).name());
            return static_cast<T *>(this);
        }

        template<typename T>
        const T *as_unchecked() const {
            static_assert(std::is_base_of_v<BaseVector, T>);
            POLLUX_DCHECK_NOT_NULL(
                dynamic_cast<const T*>(this),
                "Wrong type cast expected {}, but got {}",
                typeid(T).name(),
                typeid(*this).name());
            return static_cast<const T *>(this);
        }

        template<typename T>
        T *as_checked() {
            auto *casted = as<T>();
            POLLUX_CHECK_NOT_NULL(
                casted,
                "Wrong type cast expected {}, but got {}",
                typeid(T).name(),
                typeid(*this).name());
            return casted;
        }

        template<typename T>
        const T *as_checked() const {
            auto *casted = as<T>();
            POLLUX_CHECK_NOT_NULL(
                casted,
                "Wrong type cast expected {}, but got {}",
                typeid(T).name(),
                typeid(*this).name());
            return casted;
        }

        template<typename T>
        const FlatVector<T> *as_flat_vector() const {
            return dynamic_cast<const FlatVector<T> *>(this);
        }

        template<typename T>
        FlatVector<T> *as_flat_vector() {
            return dynamic_cast<FlatVector<T> *>(this);
        }

        pollux::memory::MemoryPool *pool() const {
            return pool_;
        }

        virtual bool is_null_at(vector_size_t idx) const {
            POLLUX_DCHECK_GE(idx, 0, "Index must not be negative");
            POLLUX_DCHECK_LT(idx, length_, "Index is too large");
            return rawNulls_ ? bits::isBitNull(rawNulls_, idx) : false;
        }

        /// Returns true if value at specified index is null or contains null.
        /// Primitive type values can be null, but cannot contain nulls. Arrays, maps
        /// and structs can be null and can contains nulls. Non-null array may contain
        /// one or more elements that are null or contain nulls themselves. Non-null
        /// maps may contain one more entry with key or value that's null or contains
        /// null. Non-null struct may contain a field that's null or contains null.
        virtual bool contains_null_at(vector_size_t idx) const = 0;

        std::optional<vector_size_t> get_null_count() const {
            return nullCount_;
        }

        void set_null_count(vector_size_t newNullCount) {
            nullCount_ = newNullCount;
        }

        const TypePtr &type() const {
            return type_;
        }

        bool type_uses_custom_comparison() const {
            return typeUsesCustomComparison_;
        }

        /// Changes vector type. The new type can have a different
        /// logical representation while maintaining the same physical type.
        /// Additionally, note that the caller must ensure that this vector is not
        /// shared, i.e. singly-referenced.
        virtual void set_type(const TypePtr &type) {
            POLLUX_CHECK_NOT_NULL(type);
            POLLUX_CHECK(
                type_->kindEquals(type),
                "Cannot change vector type from {} to {}. The old and new types can be different logical types, but the underlying physical types must match.",
                type_,
                type);
            type_ = type;
        }

        TypeKind type_kind() const {
            return typeKind_;
        }

        /// Returns a smart pointer to the null bitmap data for this vector. May hold
        /// nullptr if there are no nulls. Not const because some vectors may generate
        /// this on first access. For ConstantVector, this method returns a BufferPtr
        /// of only size 1. For DictionaryVector, this method returns a BufferPtr for
        /// only nulls in the top-level layer.
        const BufferPtr &nulls() const {
            return nulls_;
        }

        /// Returns a pointer to the raw null bitmap buffer of this vector. Notice
        /// that users should not used this API to access nulls directly of a
        /// ConstantVector or DictionaryVector. If the vector is a ConstantVector,
        /// rawNulls_ is only of size 1. If the vector is a DictionaryVector,
        /// rawNulls_ points to a raw buffer of only nulls in the top-level layer.
        /// Nulls of a ConstantVector or DictionaryVector can be accessed through the
        /// is_null_at() API or DecodedVector.
        const uint64_t *raw_nulls() const {
            return rawNulls_;
        }

        /// Ensures that nulls are writable (mutable and single referenced for
        /// BaseVector::length_).
        uint64_t *mutable_raw_nulls() {
            ensure_nulls();
            return const_cast<uint64_t *>(rawNulls_);
        }

        /// Ensures the vector has capacity for the nulls and returns the shared
        /// pointer to the buffer containing them.
        /// Optional parameter 'setNotNull' is passed to ensure_nulls_capacity() and is
        /// used to ensure all the rows will be 'not nulls' if set to true.
        BufferPtr &mutable_nulls(vector_size_t size, bool setNotNull = false) {
            ensure_nulls_capacity(size, setNotNull);
            return nulls_;
        }

        std::optional<vector_size_t> get_distinct_value_count() const {
            return distinctValueCount_;
        }

        /// @return the number of rows of data in this vector
        vector_size_t size() const {
            return length_;
        }

        virtual void append(const BaseVector *other) {
            auto totalSize = BaseVector::length_ + other->size();
            auto previousSize = BaseVector::size();
            resize(totalSize);
            copy(other, previousSize, 0, other->size());
        }

        /// @return the number of bytes this vector takes on disk when in a compressed
        /// and serialized format
        std::optional<ByteCount> storage_bytes() const {
            return storageByteCount_;
        }

        /// @return the number of bytes required to naively represent all of the data
        /// in this vector - the raw data size if not in a compressed or otherwise
        /// optimized format
        std::optional<ByteCount> represented_bytes() const {
            return representedByteCount_;
        }

        /// @return the number of bytes required to hold this vector in memory
        ByteCount in_memory_bytes() const {
            return in_memory_bytes_;
        }

        /// @return true if this vector has the same value at the given index as the
        /// other vector at the other vector's index (including if both are null),
        /// false otherwise
        /// @throws if the type_ of other doesn't match the type_ of this
        virtual bool equal_value_at(
            const BaseVector *other,
            vector_size_t index,
            vector_size_t otherIndex) const {
            static constexpr CompareFlags kEqualValueAtFlags =
                    CompareFlags::equality(CompareFlags::NullHandlingMode::kNullAsValue);

            // Will always have value because nullHandlingMode is NullAsValue.
            return compare(other, index, otherIndex, kEqualValueAtFlags).value() == 0;
        }

        /// Returns true if this vector has the same value at the given index as the
        /// other vector at the other vector's index (including if both are null when
        /// nullHandlingMode is NullAsValue), false otherwise. If nullHandlingMode is
        /// StopAtNull, returns std::nullopt if null encountered.
        virtual std::optional<bool> equal_value_at(
            const BaseVector *other,
            vector_size_t index,
            vector_size_t otherIndex,
            CompareFlags::NullHandlingMode nullHandlingMode) const;

        int32_t compare(
            const BaseVector *other,
            vector_size_t index,
            vector_size_t otherIndex) const {
            // Default compare flags always generate value.
            return compare(other, index, otherIndex, CompareFlags()).value();
        }

        /// When CompareFlags is ASCENDING, returns < 0 if 'this' at 'index' is less
        /// than 'other' at 'otherIndex', 0 if equal and > 0 otherwise.
        /// When CompareFlags is DESCENDING, returns < 0 if 'this' at 'index' is
        /// larger than 'other' at 'otherIndex', 0 if equal and < 0 otherwise. If
        /// flags.nullHandlingMode is not NullAsValue, the function may returns
        /// std::nullopt if null encountered.
        virtual std::optional<int32_t> compare(
            const BaseVector *other,
            vector_size_t index,
            vector_size_t otherIndex,
            CompareFlags flags) const = 0;

        /// Sort values at specified 'indices'. Used to sort map keys.
        virtual void sort_indices(
            std::vector<vector_size_t> &indices,
            CompareFlags flags) const {
            std::sort(
                indices.begin(),
                indices.end(),
                [&](vector_size_t left, vector_size_t right) {
                    return compare(this, left, right, flags) < 0;
                });
        }

        /// Sort values at specified 'indices' after applying the 'mapping'. Used to
        /// sort map keys.
        virtual void sort_indices(
            std::vector<vector_size_t> &indices,
            const vector_size_t *mapping,
            CompareFlags flags) const {
            std::sort(
                indices.begin(),
                indices.end(),
                [&](vector_size_t left, vector_size_t right) {
                    return compare(this, mapping[left], mapping[right], flags) < 0;
                });
        }

        /// Compares values in range [start, start + size) and returns an index of a
        /// duplicate value if found.
        std::optional<vector_size_t> find_duplicate_value(
            vector_size_t start,
            vector_size_t size,
            CompareFlags flags);

        /// @return the hash of the value at the given index in this vector
        virtual uint64_t hash_value_at(vector_size_t index) const = 0;

        /// @return a new vector that contains the hashes for all entries
        virtual std::unique_ptr<SimpleVector<uint64_t> > hash_all() const = 0;

        /// Returns true if this vector is encoded as flat (FlatVector).
        bool is_flat_encoding() const {
            return encoding_ == VectorEncoding::Simple::FLAT;
        }

        /// Returns true if this vector is encoded as constant (ConstantVector).
        bool is_constant_encoding() const {
            return encoding_ == VectorEncoding::Simple::CONSTANT;
        }

        /// Returns true if this vector has a scalar type. If so, values are
        /// accessed by value_at after casting the vector to a type()
        /// dependent instantiation of SimpleVector<T>.
        virtual bool is_scalar() const {
            return false;
        }

        /// Returns the scalar or complex vector wrapped inside any nesting of
        /// dictionary, sequence or constant vectors.
        virtual const BaseVector *wrapped_vector() const {
            return this;
        }

        static const VectorPtr &wrapped_vector_shared(const VectorPtr &vector);

        /// Returns the index to apply for 'index' in the vector returned by
        /// wrapped_vector(). Translates the index over any nesting of dictionaries,
        /// sequences and constants.
        virtual vector_size_t wrapped_index(vector_size_t index) const {
            return index;
        }

        /// Sets the null indicator at 'idx'. This API throws if the vector is a
        /// ConstantVector.
        MELON_ALWAYS_INLINE virtual void set_null(vector_size_t idx, bool isNull) {
            POLLUX_DCHECK(idx >= 0 && idx < length_);
            if (!nulls_ && !isNull) {
                return;
            }
            ensure_nulls();
            bits::set_null(nulls_->asMutable<uint64_t>(), idx, isNull);
        }

        struct CopyRange {
            vector_size_t sourceIndex;
            vector_size_t targetIndex;
            vector_size_t count;
        };

        /// Sets null flags for each row in 'ranges' to 'isNull'.
        static void set_nulls(
            uint64_t *rawNulls,
            const melon::Range<const CopyRange *> &ranges,
            bool isNull);

        /// Copies null flags for each row in 'ranges' from 'sourceRawNulls' to
        /// 'targetRawNulls'.
        static void copy_nulls(
            uint64_t *targetRawNulls,
            const uint64_t *sourceRawNulls,
            const melon::Range<const CopyRange *> &ranges);

        static int32_t
        count_nulls(const BufferPtr &nulls, vector_size_t begin, vector_size_t end) {
            return nulls ? bits::count_nulls(nulls->as<uint64_t>(), begin, end) : 0;
        }

        static int32_t count_nulls(const BufferPtr &nulls, vector_size_t size) {
            return count_nulls(nulls, 0, size);
        }

        /// Returns whether or not the nulls buffer can be modified.
        /// This does not guarantee the existence of the nulls buffer, if using this
        /// within BaseVector you still may need to call ensure_nulls.
        virtual bool is_nulls_writable() const {
            return !nulls_ || (nulls_->isMutable());
        }

        /// Sets null when 'nulls' has a null value for active rows in 'rows'.
        /// Is a no-op 'nulls' is a nullptr or 'rows' has no selections. This API
        /// throws if the vector is a ConstantVector.
        virtual void add_nulls(const uint64_t *nulls, const SelectivityVector &rows);

        /// Sets nulls for all active row in 'nullRows'. Is a no-op if nullRows has no
        /// selections. This API throws if the vector is a ConstantVector.
        virtual void add_nulls(const SelectivityVector &nullRows);

        /// Clears nulls for all active rows in 'nonNullRows'
        virtual void clear_nulls(const SelectivityVector &nonNullRows);

        /// Clears nulls for all row indices in range [begin, end).
        virtual void clear_nulls(vector_size_t begin, vector_size_t end);

        void clear_all_nulls() {
            clear_nulls(0, size());
        }

        void reuseNulls();

        /// Sets the size to 'newSize' and ensures there is space for the
        /// indicated number of nulls and top level values (eg. values for Flat,
        /// indices for Dictionary, etc). Any immutable buffers that need to be
        /// resized are copied. 'setNotNull' indicates if nulls in range [oldSize,
        /// newSize] should be set to not null. Note: caller must ensure that the
        /// vector is writable; for instance have recursively single referenced
        /// buffers and vectors.
        virtual void resize(vector_size_t newSize, bool setNotNull = true);

        /// Sets the rows of 'this' given by 'rows' to
        /// 'source.value_at(toSourceRow ? toSourceRow[row] : row)', where
        /// 'row' iterates over 'rows'. All active 'row' in 'rows' must map to a valid
        /// row in the 'source'.
        virtual void copy(
            const BaseVector *source,
            const SelectivityVector &rows,
            const vector_size_t *toSourceRow);

        /// Utility for making a deep copy of a whole vector.
        static VectorPtr copy(
            const BaseVector &vector,
            pollux::memory::MemoryPool *pool = nullptr) {
            auto result = BaseVector::create(
                vector.type(), vector.size(), pool ? pool : vector.pool());
            result->copy(&vector, 0, 0, vector.size());
            return result;
        }

        virtual void copy(
            const BaseVector *source,
            vector_size_t targetIndex,
            vector_size_t sourceIndex,
            vector_size_t count) {
            if (count == 0) {
                return;
            }
            CopyRange range{sourceIndex, targetIndex, count};
            copy_ranges(source, melon::Range(&range, 1));
        }

        /// Converts SelectivityVector into a list of CopyRanges having sourceIndex ==
        /// targetIndex. Aims to produce as few ranges as possible. If all rows are
        /// selected, returns a single range.
        static std::vector<CopyRange> to_copy_ranges(const SelectivityVector &rows);

        /// Copy multiple ranges at once.  This is more efficient than calling `copy`
        /// multiple times, especially for ARRAY, MAP, and VARCHAR.
        virtual void copy_ranges(
            const BaseVector * /*source*/,
            const melon::Range<const CopyRange *> & /*ranges*/) {
            POLLUX_UNSUPPORTED("Can only copy into flat or complex vectors");
        }

        /// This makes a deep copy of the Vector allocating new child Vectors and
        /// Buffers recursively.  Unlike copy, this preserves encodings recursively.
        virtual VectorPtr copy_preserve_encodings(
            pollux::memory::MemoryPool *pool = nullptr) const = 0;

        /// Construct a zero-copy slice of the vector with the indicated offset and
        /// length.
        virtual VectorPtr slice(vector_size_t offset, vector_size_t length) const = 0;

        /// Transposes two sets of dictionary indices into one level of indirection.
        /// Sets result[i] = base[indices[i]] for i = 0 ... i < size.
        static void transpose_indices(
            const vector_size_t *base,
            vector_size_t size,
            const vector_size_t *indices,
            vector_size_t *result);

        /// Transposes two levels of indices into a single level with nulls. sets
        /// result[i] = base[indices[i]] where i is not null in 'wrapNulls' and
        /// indices[i] is not null in 'baseNulls'. If indices[i] is null in
        /// 'baseNulls' or i is null in 'wrapNulls', then 'resultNulls' is null at i.
        /// 'wrapNulls' may be nullptr, meaning that no new nulls are added.
        static void transpose_indices_with_nulls(
            const vector_size_t *baseIndices,
            const uint64_t *baseNulls,
            vector_size_t wrapSize,
            const vector_size_t *wrapIndices,
            const uint64_t *wrapNulls,
            vector_size_t *resultIndices,
            uint64_t *resultNulls);

        /// Flattens 'dictionaryValues', which is a dictionary and replaces
        /// it with its base. 'size' is the number of valid elements in
        /// 'indices' and 'nulls'. Null positions may have an invalid
        /// index. Rewrites 'indices' from being indices into
        /// 'dictionaryValues' to being indices into the latter's
        /// base. Rewrites 'nulls' to be nulls from 'dictionaryValues' and
        /// its base vector. This is used when a dictionary vector loads a
        /// lazy values vector and finds out that the loaded is itself a
        /// dictionary.
        static void transpose_dictionary_values(
            vector_size_t wrapSize,
            BufferPtr &wrapNulls,
            BufferPtr &wrapIndices,
            std::shared_ptr<BaseVector> &dictionaryValues);

        // Returns a vector of the type of 'source' where 'indices' contains
        // an index into 'source' for each element of 'source'. The
        // resulting vector has position i set to source[i]. This is
        // equivalent to wrapping 'source' in a dictionary with 'indices'
        // but this may reuse structure if said structure is uniquely owned
        // or if a copy is more efficient than dictionary wrapping.
        static VectorPtr transpose(BufferPtr indices, VectorPtr &&source);

        static VectorPtr create_constant(
            const TypePtr &type,
            variant value,
            vector_size_t size,
            pollux::memory::MemoryPool *pool);

        static VectorPtr create_null_constant(
            const TypePtr &type,
            vector_size_t size,
            pollux::memory::MemoryPool *pool);

        /// Wraps the 'vector' in the provided dictionary encoding. If the input
        /// vector is constant and the nulls buffer is empty, this method may return a
        /// ConstantVector. Additionally, if 'flattenIfRedundant' is true, this method
        /// may return a flattened version of the expected dictionary vector if
        /// applying the dictionary encoding would result in a suboptimally encoded
        /// vector.
        static VectorPtr wrap_in_dictionary(
            BufferPtr nulls,
            BufferPtr indices,
            vector_size_t size,
            VectorPtr vector,
            bool flattenIfRedundant = false);

        static VectorPtr
        wrap_in_sequence(BufferPtr lengths, vector_size_t size, VectorPtr vector);

        /// Creates a ConstantVector of specified length and value coming from the
        /// 'index' element of the 'vector'. Peels off any encodings of the 'vector'
        /// before making a new ConstantVector. The result vector is either a
        /// ConstantVector holding a scalar value or a ConstantVector wrapping flat or
        /// lazy vector. The result cannot be a wrapping over another constant or
        /// dictionary vector. If copyBase is true and the result vector wraps a
        /// vector, the wrapped vector is newly constructed by copying the value from
        /// the original, guaranteeing no Vectors are shared with 'vector'.
        static VectorPtr wrap_in_constant(
            vector_size_t length,
            vector_size_t index,
            VectorPtr vector,
            bool copyBase = false);

        /// Makes 'result' writable for 'rows'. A wrapper (e.g. dictionary, constant,
        /// sequence) is flattened and a multiply referenced flat vector is copied.
        /// The content of 'rows' is not copied, as these values are intended to be
        /// overwritten.
        ///
        /// After invoking this function, the 'result' is guaranteed to be a flat
        /// uniquely-referenced vector with all data-dependent flags reset.
        ///
        /// Use SelectivityVector::empty() to make the 'result' writable and preserve
        /// all current values.
        ///
        /// If 'result' is a lazy vector, then caller needs to ensure it is unique in
        /// order to re-use the loaded vector. Otherwise, a copy would be created.
        static void ensure_writable(
            const SelectivityVector &rows,
            const TypePtr &type,
            pollux::memory::MemoryPool *pool,
            VectorPtr &result,
            VectorPool *vectorPool = nullptr);

        virtual void ensure_writable(const SelectivityVector &rows);

        /// Returns true if the following conditions hold:
        ///  * The vector is singly referenced.
        ///  * The vector has a Flat-like encoding (Flat, Array, Map, Row).
        ///  * Any child Buffers are mutable  and singly referenced.
        ///  * All of these conditions hold for child Vectors recursively.
        /// This function is templated rather than taking a
        /// std::shared_ptr<BaseVector> because if we were to do that the compiler
        /// would allocate a new shared_ptr when this function is called making it not
        /// unique.
        template<typename T>
        static bool is_vector_writable(const std::shared_ptr<T> &vector) {
            if (vector.use_count() != 1) {
                return false;
            }

            return vector->is_writable();
        }

        virtual bool is_writable() const {
            return false;
        }

        /// If 'vector' consists of a single value and is longer than one,
        /// returns an equivalent constant vector, else nullptr.
        static VectorPtr constantify(
            const std::shared_ptr<BaseVector> &vector,
            DecodedVector *decoded = nullptr);

        /// Flattens the input vector and all of its children.
        static void flattenVector(VectorPtr &vector);

        template<typename T>
        static inline uint64_t byteSize(vector_size_t count) {
            return sizeof(T) * count;
        }

        /// If 'vector' is a wrapper, returns the underlying values vector. This is
        /// virtual and defined here because we must be able to access this in type
        /// agnostic code without a switch on all data types.
        virtual const VectorPtr &value_vector() const {
            POLLUX_UNSUPPORTED("Vector is not a wrapper");
        }

        virtual void set_value_vector(VectorPtr v) {
            POLLUX_UNSUPPORTED("Vector is not a wrapper");
        }

        virtual BaseVector *loaded_vector() {
            return this;
        }

        virtual const BaseVector *loaded_vector() const {
            return this;
        }

        static const VectorPtr &loaded_vector_shared(const VectorPtr &vector);

        virtual const BufferPtr &values() const {
            POLLUX_UNSUPPORTED("Only flat vectors have a values buffer");
        }

        virtual const void *values_as_void() const {
            POLLUX_UNSUPPORTED("Only flat vectors have a values buffer");
        }

        /// If 'this' is a wrapper, returns the wrap info, interpretation depends on
        /// encoding.
        virtual const BufferPtr &wrap_info() const {
            POLLUX_UNSUPPORTED("Vector is not a wrapper");
        }

        virtual void setWrapInfo(BufferPtr wrap_info) {
            POLLUX_UNSUPPORTED("Vector is not a wrapper");
        }

        template<typename T = BaseVector>
        static std::shared_ptr<T> create(
            const TypePtr &type,
            vector_size_t size,
            pollux::memory::MemoryPool *pool) {
            return std::static_pointer_cast<T>(create_internal(type, size, pool));
        }

        static VectorPtr get_or_create_empty(
            VectorPtr vector,
            const TypePtr &type,
            pollux::memory::MemoryPool *pool) {
            return vector ? vector : create(type, 0, pool);
        }

        /// Set 'nulls' to be the nulls buffer of this vector. This API should not be
        /// used on ConstantVector.
        void set_nulls(const BufferPtr &nulls);

        /// Reset the nulls buffer of this vector to be empty. This API should not be
        /// used on ConstantVector.
        void reset_nulls() {
            set_nulls(nullptr);
        }

        /// Ensures that 'indices' is singly-referenced and has space for 'newSize'
        /// elements. Sets elements between the 'currentSize' and 'newSize' to 0 if
        /// 'newSize' > 'currentSize'.
        ///
        /// If 'indices' is nullptr, read-only, not uniquely-referenced, or doesn't
        /// have capacity for 'newSize' elements allocates new buffer and copies data
        /// to it. Updates '*rawIndices' to point to the start of 'indices' buffer.
        static void resize_indices(
            vector_size_t currentSize,
            vector_size_t newSize,
            pollux::memory::MemoryPool *pool,
            BufferPtr &indices,
            const vector_size_t **rawIndices);

        /// Makes sure '*buffer' has space for 'size' items of T and is writable. Sets
        /// 'raw' to point to the writable contents of '*buffer'.
        template<typename T, typename RawT>
        static void ensure_buffer(
            vector_size_t size,
            pollux::memory::MemoryPool *pool,
            BufferPtr *buffer,
            RawT **raw) {
            vector_size_t minBytes = byteSize<T>(size);
            if (*buffer && (*buffer)->capacity() >= minBytes && (*buffer)->unique()) {
                (*buffer)->setSize(minBytes);
                if (raw) {
                    *raw = (*buffer)->asMutable<RawT>();
                }
                return;
            }
            if (*buffer) {
                AlignedBuffer::reallocate<T>(buffer, size);
            } else {
                *buffer = AlignedBuffer::allocate<T>(size, pool);
            }
            if (raw) {
                *raw = (*buffer)->asMutable<RawT>();
            }
            (*buffer)->setSize(minBytes);
        }

        /// Returns the byte size of memory that is kept live through 'this'.
        virtual uint64_t retained_size() const {
            return nulls_ ? nulls_->capacity() : 0;
        }

        /// Returns an estimate of the 'retained_size' of a flat representation of the
        /// data stored in this vector. Returns zero if this is a lazy vector that
        /// hasn't been loaded yet.
        virtual uint64_t estimate_flat_size() const;

        /// To safely reuse a vector one needs to (1) ensure that the vector as well
        /// as all its buffers and child vectors are singly-referenced and mutable
        /// (for buffers); (2) clear append-only string buffers and child vectors
        /// (elements of arrays, keys and values of maps, fields of structs); (3)
        /// reset all data-dependent flags.
        ///
        /// This method takes a non-const reference to a 'vector' and updates it to
        /// possibly a new flat vector of the specified size that is safe to reuse.
        /// If input 'vector' is not singly-referenced or not flat, replaces 'vector'
        /// with a new vector of the same type and specified size. If some of the
        /// buffers cannot be reused, these buffers are reset. Child vectors are
        /// updated by calling this method recursively with size zero. Data-dependent
        /// flags are reset after this call.
        static void prepare_for_reuse(VectorPtr &vector, vector_size_t size);

        /// Resets non-reusable buffers and updates child vectors by calling
        /// BaseVector::prepare_for_reuse.
        /// Base implementation checks and resets nulls buffer if needed. Keeps the
        /// nulls buffer if singly-referenced, mutable and has at least one null bit
        /// set.
        virtual void prepare_for_reuse();

        /// Returns a brief summary of the vector. If 'recursive' is true, includes a
        /// summary of all the layers of encodings starting with the top layer.
        ///
        /// For example,
        ///     with recursive 'false':
        ///
        ///         [DICTIONARY INTEGER: 5 elements, no nulls]
        ///
        ///     with recursive 'true':
        ///
        ///         [DICTIONARY INTEGER: 5 elements, no nulls], [FLAT INTEGER: 10
        ///             elements, no nulls]
        std::string toString(bool recursive) const;

        /// Same as toString(false). Provided to allow for easy invocation from LLDB.
        std::string toString() const {
            return toString(false);
        }

        /// Returns string representation of the value in the specified row.
        virtual std::string toString(vector_size_t index) const;

        /// Returns a list of values in rows [from, to).
        ///
        /// Automatically adjusts 'from' and 'to' to a range of valid indices. Returns
        /// empty string if 'from' is greater than or equal to vector size or 'to' is
        /// less than or equal to zero. Returns values up to the end of the vector if
        /// 'to' is greater than vector size. Returns values from the start of the
        /// vector if 'from' is negative.
        ///
        /// The type of the 'delimiter' is a const char* and not an std::string to
        /// allow for invoking this method from LLDB.
        std::string toString(
            vector_size_t from,
            vector_size_t to,
            const char *delimiter,
            bool includeRowNumbers = true) const;

        /// Returns a list of values in rows [from, to). Values are separated by a new
        /// line and prefixed with a row number.
        ///
        /// This method is provided to allow to easy invocation from LLDB.
        std::string toString(vector_size_t from, vector_size_t to) const {
            return toString(from, to, "\n");
        }

        /// Marks the vector as containing or being a lazy vector and being wrapped.
        /// Should only be used if 'this' is lazy or has a nested lazy vector.
        /// Returns true if this is the first time it was wrapped, else returns false.
        bool mark_as_containing_lazy_and_wrapped() {
            if (containsLazyAndIsWrapped_) {
                return false;
            }
            containsLazyAndIsWrapped_ = true;
            return true;
        }

        void clear_containing_lazy_and_wrapped() {
            containsLazyAndIsWrapped_ = false;
        }

        bool memo_disabled() const {
            return memoDisabled_;
        }

        void disable_memo() {
            memoDisabled_ = true;
        }

        /// Used to check internal state of a vector like sizes of the buffers,
        /// enclosed child vectors, values in indices. Currently, its only used in
        /// debug builds to check the result of expressions and some interim results.
        virtual void validate(const VectorValidateOptions &options = {}) const;

        MELON_ALWAYS_INLINE static std::optional<int32_t>
        compare_nulls(bool thisNull, bool otherNull, CompareFlags flags) {
            DKCHECK(thisNull || otherNull);
            switch (flags.nullHandlingMode) {
                case CompareFlags::NullHandlingMode::kNullAsIndeterminate:
                    if (flags.equalsOnly) {
                        return kIndeterminate;
                    } else {
                        POLLUX_USER_FAIL("Ordering nulls is not supported");
                    }
                case CompareFlags::NullHandlingMode::kNullAsValue:
                    if (thisNull && otherNull) {
                        return 0;
                    }

                    if (flags.nullsFirst) {
                        return thisNull ? -1 : 1;
                    }
                    return thisNull ? 1 : -1;
            }

            POLLUX_UNREACHABLE(
                "The function should be called only if one of the inputs is null");
        }

        /// Reset data-dependent flags to the "unknown" status. This is needed
        /// whenever a vector is mutated because the modification may invalidate these
        /// flags. Currently, we call this function in BaseVector::ensure_writable()
        /// and BaseVector::prepare_for_reuse() that are expected to be called before
        /// any vector mutation.
        ///
        /// Per-vector flags are reset to default values. Per-row flags are reset only
        /// at the selected rows. If rows is a nullptr, per-row flags are reset at all
        /// rows.
        virtual void reset_data_dependent_flags(const SelectivityVector * /*rows*/) {
            nullCount_ = std::nullopt;
            distinctValueCount_ = std::nullopt;
            representedByteCount_ = std::nullopt;
            storageByteCount_ = std::nullopt;
        }

    protected:
        // Returns a brief summary of the vector. The default implementation includes
        // encoding, type, number of rows and number of nulls.
        //
        // For example,
        //     [FLAT INTEGER: 3 elements, no nulls]
        //     [DICTIONARY INTEGER: 5 elements, 1 nulls]
        virtual std::string to_summary_string() const;

        // Allocates or reallocates nulls_ with at least the given size if nulls_
        // hasn't been allocated yet or has been allocated with a smaller capacity.
        // Ensures that nulls are writable (mutable and single referenced for
        // minimumSize).
        void ensure_nulls_capacity(vector_size_t minimumSize, bool setNotNull = false);

        void ensure_nulls() {
            ensure_nulls_capacity(length_, true);
        }

        BufferPtr slice_nulls(vector_size_t offset, vector_size_t length) const {
            return nulls_ ? Buffer::slice<bool>(nulls_, offset, length, pool_) : nulls_;
        }

        TypePtr type_;
        const TypeKind typeKind_;
        // Whether `type_` is a type that provides custom comparison operations.
        // We use this instead of calling type_->providesCustomCompare() because
        // having a constant field helps the compiler to pull this condition up in
        // loops, and `type_` itself is non-constant (though it can only be modified
        // logically, so this property is safe to store).
        const bool typeUsesCustomComparison_;
        const VectorEncoding::Simple encoding_;
        BufferPtr nulls_;
        // Caches raw pointer to 'nulls->as<uint64_t>().
        const uint64_t *rawNulls_ = nullptr;
        pollux::memory::MemoryPool *pool_;
        tsan_atomic<vector_size_t> length_{0};

        // Holds the number of nulls in the vector. If the number of nulls is not
        // available, it is set to std::nullopt. Setting the value to zero does have
        // implications (SIMD operations need null count to be zero) and is not the
        // same as std::nullopt.
        std::optional<vector_size_t> nullCount_;
        std::optional<vector_size_t> distinctValueCount_;
        std::optional<ByteCount> representedByteCount_;
        std::optional<ByteCount> storageByteCount_;
        ByteCount in_memory_bytes_ = 0;

    private:
        static VectorPtr create_internal(
            const TypePtr &type,
            vector_size_t size,
            pollux::memory::MemoryPool *pool);

        friend class LazyVector;

        // Is true if this vector is a lazy vector or contains one and is being
        // wrapped. Keeping track of this helps to enforce the invariant that an
        // unloaded lazy vector should not be wrapped by two separate top level
        // vectors. This would ensure we avoid it being loaded for two separate set
        // of rows.
        std::atomic_bool containsLazyAndIsWrapped_{false};

        // Whether we should use Expr::evalWithMemo to cache the result of evaluation
        // on dictionary values (this vector).  Set to false when the dictionary
        // values are not going to be reused (e.g. result of filtering), so that we
        // don't need to reallocate the result for every batch.
        bool memoDisabled_{false};
    };

    /// Loops over rows in 'ranges' and invokes 'func' for each row.
    /// @param func A void function taking two arguments: targetIndex and
    /// sourceIndex.
    template<typename TFunc>
    void apply_to_each_row(
        const melon::Range<const BaseVector::CopyRange *> &ranges,
        const TFunc &func) {
        for (const auto &range: ranges) {
            for (auto i = 0; i < range.count; ++i) {
                func(range.targetIndex + i, range.sourceIndex + i);
            }
        }
    }

    /// Loops over 'ranges' and invokes 'func' for each range.
    /// @param func A void function taking 3 arguments: targetIndex, sourceIndex
    /// and count.
    template<typename TFunc>
    void apply_to_each_range(
        const melon::Range<const BaseVector::CopyRange *> &ranges,
        const TFunc &func) {
        for (const auto &range: ranges) {
            func(range.targetIndex, range.sourceIndex, range.count);
        }
    }

    template<>
    uint64_t BaseVector::byteSize<bool>(vector_size_t count);

    template<>
    inline uint64_t BaseVector::byteSize<UnknownValue>(vector_size_t) {
        return 0;
    }

    /// Returns true if vector is a Lazy vector, possibly wrapped, that hasn't
    /// been loaded yet.
    bool is_lazy_not_loaded(const BaseVector &vector);

    /// Allocates a buffer to fit at least 'size' indices and initializes them to
    /// zero.
    inline BufferPtr allocate_indices(vector_size_t size, memory::MemoryPool *pool) {
        return AlignedBuffer::allocate<vector_size_t>(size, pool, 0);
    }

    /// Allocates a buffer to fit at least 'size' null bits and initializes them to
    /// the provided 'initValue' which has a default value of non-null.
    inline BufferPtr allocate_nulls(
        vector_size_t size,
        memory::MemoryPool *pool,
        bool initValue = bits::kNotNull) {
        return AlignedBuffer::allocate<bool>(size, pool, initValue);
    }

    /// Returns a summary of the null bits in the specified buffer and prints out
    /// first 'maxBitsToPrint' bits. Automatically adjusts if 'maxBitsToPrint' is
    /// greater than total number of bits available.
    /// For example: 3 out of 8 rows are null: .nn.n...
    std::string print_nulls(
        const BufferPtr &nulls,
        vector_size_t maxBitsToPrint = 30);

    /// Returns a summary of the indices buffer and prints out first
    /// 'maxIndicesToPrint' indices. Automatically adjusts if 'maxIndicesToPrint' is
    /// greater than total number of indices available.
    /// For example: 5 unique indices out of 6: 34, 79, 11, 0, 0, 33.
    std::string print_indices(
        const BufferPtr &indices,
        vector_size_t maxIndicesToPrint = 10);

    template<typename OutputStream>
    OutputStream &operator<<(
        OutputStream &out,
        const BaseVector::CopyRange &range) {
        out << "{sourceIndex=" << range.sourceIndex
                << " targetIndex=" << range.targetIndex << " count=" << range.count
                << "}";
        return out;
    }
} // namespace kumo::pollux

namespace melon {
    /// Allow VectorEncoding::Simple to be transparently used by melon::sformat.
    /// e.g: melon::sformat("type: {}", encodingType);
    template<>
    class FormatValue<kumo::pollux::VectorEncoding::Simple> {
    public:
        explicit FormatValue(const kumo::pollux::VectorEncoding::Simple &type)
            : type_(type) {
        }

        template<typename FormatCallback>
        void format(FormatArg &arg, FormatCallback &cb) const {
            return format_value::formatString(
                kumo::pollux::VectorEncoding::mapSimpleToName(type_), arg, cb);
        }

    private:
        kumo::pollux::VectorEncoding::Simple type_;
    };
} // namespace melon

template<>
struct fmt::formatter<kumo::pollux::VectorEncoding::Simple> {
    constexpr auto parse(format_parse_context &ctx) {
        return ctx.begin();
    }

    template<typename FormatContext>
    auto format(
        const kumo::pollux::VectorEncoding::Simple &x,
        FormatContext &ctx) const {
        return fmt::format_to(
            ctx.out(), "{}", kumo::pollux::VectorEncoding::mapSimpleToName(x));
    }
};
